package com.shujia.dim

import com.shujia.common.SparkTool
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object DimScenicGrid extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 加载景区边界表以及网格配置表
    val scenicDF: DataFrame = spark.table("dim.dim_scenic_boundary")
    val gridDF: DataFrame = spark.table("dim.dim_geotag_grid")

    // 将两份数据进行笛卡尔积操作，用于比较
    // 因为两张表的数据变化都不是很大，故可做笛卡尔积，做一次后续就可以多次使用

    gridDF
      .join(scenicDF)
      .where(isInBoundaryWithLgLat($"boundary", $"center_longi", $"center_lati"))
      // 按照景区id、景区名称进行分组
      .groupBy($"scenic_id", $"scenic_name")
      .agg(concat_ws("|", collect_set($"grid_id")) as "grids")
      .write
      .format("csv")
      .option("sep", ",")
      .mode(SaveMode.Overwrite)
      .save("/daas/motl/dim/dim_scenic_grid")

  }
}
